package org.example.api.window;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.example.api.bean.SensorReading;

/**
 * @author huangqihan
 * @date 2021/2/22
 */
public class CountWindowTest {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String host = "localhost";
        int port = 7777;

        // Use NetCat to read the data source. nc -lp 7777
        DataStreamSource<String> inputStream = env.socketTextStream(host, port);

        SingleOutputStreamOperator<SensorReading> mapStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // CountWindow functions
        DataStream<Double> resultStream = mapStream.keyBy("id")
                .countWindow(10, 2)
                .aggregate(
                        new AggregateFunction<SensorReading, Tuple2<Double, Integer>, Double>() {

                            @Override
                            public Tuple2<Double, Integer> createAccumulator() {
                                return new Tuple2<>(0.0, 0);
                            }

                            @Override
                            public Tuple2<Double, Integer> add(SensorReading sensorReading, Tuple2<Double, Integer> acc) {
                                return new Tuple2<>(acc.f0 + sensorReading.getTemperature(), acc.f1 + 1);
                            }

                            @Override
                            public Double getResult(Tuple2<Double, Integer> acc) {
                                return acc.f0 / acc.f1;
                            }

                            @Override
                            public Tuple2<Double, Integer> merge(Tuple2<Double, Integer> a, Tuple2<Double, Integer> b) {
                                return new Tuple2<>(a.f0 + b.f0, b.f1 + b.f1);
                            }
                        });

        resultStream.print();
        env.execute();
    }
}
